package day02;

import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.aggregation.AggregationFunction;
import org.apache.flink.util.Collector;


/**
 * 聚合类算子的使用
 */
public class AggragateFunctionTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //socket  (hellow 1)
        DataStreamSource<String> ds = env.socketTextStream("192.168.21.159", 6667);
        ds.map(new MapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                String[] arr = s.split(" ");
                return Tuple2.of(arr[0],Integer.valueOf(arr[1]));
            }
        }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                return stringIntegerTuple2.f0.substring(0,1);
            }
        })
                /**
                 *  process会将按keyBy 分到一个solt槽中所有的value都相加
                 *
                 *  process只能处理同一分区接触到的数据
                 */
                .process(new KeyedProcessFunction<String, Tuple2<String, Integer>,Tuple2<String, Integer> >() {//key (key,value) out
            int sum=0;
            @Override
            public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
                sum+=value.f1;
                out.collect(Tuple2.of(value.f0,sum));
            }
        }).print();




//                .fold(Tuple2.of("",0),new FoldFunction<Tuple2<String, Integer>,Tuple2<String, Integer>>() {
//            @Override//accumulator 初始化值
//            public Tuple2<String, Integer> fold(Tuple2<String, Integer> accumulator, Tuple2<String, Integer> value) throws Exception {
//                return Tuple2.of(value.f0,accumulator.f1+value.f1);
//            }
//        }).print();


                //AggregationFunction 和 ReduceFunction 没有区别
//                .reduce(new AggregationFunction<Tuple2<String, Integer>>() {
//            @Override
//            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> t1, Tuple2<String, Integer> t2) throws Exception {
//                return Tuple2.of(t1.f0,t1.f1+t2.f1);
//            }
//        }).print();

//                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
//            @Override
//            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> t1, Tuple2<String, Integer> t2) throws Exception {
//                return Tuple2.of(t1.f0,t1.f1+t2.f1);
//            }
//        }).print();

               // .maxBy(1).print();//max 和maxBy 区别：max只返回第一个值的key和 value 而maxBy都在乎

        env.execute();
    }
}
